package code.example.zookeeper.master;

import com.google.common.base.Throwables;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;

import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * Created by yanlei on 2017/6/9.
 */
public class ZookeeperClient {

    private ZookpeerConfig zookpeerConfig;

    public ZookpeerConfig getZookpeerConfig() {
        return zookpeerConfig;
    }

    public void setZookpeerConfig(ZookpeerConfig zookpeerConfig) {
        this.zookpeerConfig = zookpeerConfig;
    }

    private   CuratorFramework client = null;

    private  CuratorFramework getClient(){
        if(client ==null){
            synchronized (this){
                if(client == null){
                    client = initClient();
                }
            }
        }
        return client;
    }
    private  CuratorFramework initClient() {
        if(zookpeerConfig == null){
             throw new RuntimeException("zookpeerConfig is null");
        }
        zookpeerConfig.validate();
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(zookpeerConfig.getSleepTimeMsBetweenRetries(), zookpeerConfig.getMaxRetries());
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(zookpeerConfig.getConnectString())
                .retryPolicy(retryPolicy)
                .sessionTimeoutMs(zookpeerConfig.getSessionTimeoutMs())
                .connectionTimeoutMs(zookpeerConfig.getConnectTimeoutMs())
                .namespace(zookpeerConfig.getNamespace())
                .build();
        client.start();
        return client;
    }
    ConcurrentLinkedQueue<LeaderLatch> leaderLatchQueue = new ConcurrentLinkedQueue();
    public void leaderLatch(final String path,final String id, final LeaderLatchListener leaderLatchListener) {
        try {
            final CuratorFramework client = this.getClient();
            final LeaderLatch leaderLatch = new LeaderLatch(client, path, id);
            leaderLatch.addListener(leaderLatchListener);
            leaderLatch.start();
            leaderLatchQueue.add(leaderLatch);
            new Thread(new Runnable() {

                public void run() {
                    try {
                    while(true){
                             leaderLatch.await();
                            if(leaderLatch.hasLeadership()){
                               break;
                            }
                    }
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            },"ZK-LeaderLatch").start();

        }catch (Throwable e){
            throw new RuntimeException(e);
        }
    }
    public void close(){
        if(client != null){
            if(client.getState()== CuratorFrameworkState.STARTED){
                CloseableUtils.closeQuietly(client);
            }
        }
        LeaderLatch leaderLatch = null;
        while(( leaderLatch =leaderLatchQueue.peek()) != null ){
            if(leaderLatch.getState() ==(LeaderLatch.State.STARTED)){
                CloseableUtils.closeQuietly(leaderLatch);
            }
        }

    }

}
